package com.kyrie.lib_sys.rocketMQ.Producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
/**
 * rocketmq生产者
 *
 * @author heefM
 * @date 2025-08-26
 */
@Service
@Slf4j
public class MessageProducerService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送同步消息（可靠但耗时长，会等待Broker的ACK）
     *
     * @param topic
     * @param messageBody
     */
    public void sendSyncMessage(String topic, Object messageBody) {
        // 构建消息，可以添加keys, tags等
        Message<Object> message = MessageBuilder.withPayload(messageBody)
                .setHeader("KEYS", "your-key") // 设置消息Key，用于查询和追踪
                .build();

        // 发送同步消息
        // send(topic:tag, message)
        org.apache.rocketmq.client.producer.SendResult sendResult = rocketMQTemplate.syncSend(topic, message);
        log.info("同步消息发送成功！MsgId: {}, Queue: {}", sendResult.getMsgId(), sendResult.getMessageQueue());
    }

    /**
     * 发送异步消息（发送完后立即返回，通过回调告知结果）
     *
     * @param topic
     * @param messageBody
     */
    public void sendAsyncMessage(String topic, String messageBody) {
        Message<String> message = MessageBuilder.withPayload(messageBody).build();

        //异步发送,要收回一个回调接口,重写两个方法
        rocketMQTemplate.asyncSend(topic, message, new org.apache.rocketmq.client.producer.SendCallback() {
            @Override
            public void onSuccess(org.apache.rocketmq.client.producer.SendResult sendResult) {
                log.info("异步消息发送成功！MsgId: {}", sendResult.getMsgId());
            }
            @Override
            public void onException(Throwable e) {
                log.error("异步消息发送失败！", e);
            }
        });
    }

    /**
     * 发送单向消息（不关心发送结果，如日志收集）
     *
     * @param topic
     * @param messageBody
     */
    public void sendOneWayMessage(String topic, String messageBody) {
        rocketMQTemplate.sendOneWay(topic, MessageBuilder.withPayload(messageBody).build());
        log.info("单向消息已发送（不保证成功）");
    }

    /**
     * 发送带(标签)Tag的消息，格式为 Topic:Tag
     *
     * @param topic
     * @param tag
     * @param messageBody
     */
    public void sendMessageWithTag(String topic, String tag, String messageBody) {
        String destination = topic + ":" + tag; // 例如 "TestTopic:TagA"
        rocketMQTemplate.syncSend(destination, messageBody);
        log.info("带Tag的消息发送成功！Destination: {}", destination);
    }


    /**
     * 发送延迟消息
     *
     * @param topic
     * @param messageBody
     * @param delayLevel
     */
    public void sendDelayMessage(String topic, String messageBody, int delayLevel) {
        Message<String> message = MessageBuilder.withPayload(messageBody).build();
        // delayLevel: 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
        // 级别 1 对应 1秒，2对应5秒，以此类推...
        org.apache.rocketmq.client.producer.SendResult sendResult = rocketMQTemplate.syncSend(topic, message, 3000, delayLevel);
        log.info("延迟消息发送成功！MsgId: {}", sendResult.getMsgId());
    }


    /**
     * 发送缓存删除消息,同步发送消息
     *
     * @author heefM
     * @date 2025-09-05
     */
    public void sendCacheDeleteMessage(String cacheKey) {
        int maxRetries = 3; // 最大重试次数
        int retryCount = 0;
        Exception lastException = null;

        while (retryCount <= maxRetries) {
            try {
                Message<String> message = MessageBuilder.withPayload(cacheKey)
                        .setHeader(MessageConst.PROPERTY_TAGS, "CACHE_DELETE_TAG")//发送消息的二级标签--主题下面的标签
                        .build();

                rocketMQTemplate.syncSend("CACHE_DELETE_TOPIC", message);//发送的主题-----发送的消息内容
                log.info("发送缓存删除MQ消息成功，key: {}", cacheKey);
                return; // 发送成功则直接返回
            } catch (Exception e) {
                lastException = e;
                retryCount++;
                if (retryCount <= maxRetries) {
                    log.warn("发送缓存删除MQ消息失败，正在进行第{}次重试，key: {}", retryCount, cacheKey, e);
                    try {
                        Thread.sleep(1000 * retryCount); // 简单的退避策略，每次等待时间递增(每次失败后再重试的时间增加一倍)
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        log.error("发送缓存删除MQ消息线程被中断，key: {}", cacheKey, ie);
                        break;
                    }
                } else {
                    log.error("发送缓存删除MQ消息最终失败，已重试{}次，key: {}", maxRetries, cacheKey, e);
                }
            }
        }
    }


}